package SQL_L

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.types.{DoubleType, IntegerType, LongType, StructField, StructType}
import org.junit.Test

class NullProcessor {
  //创建SparkSession
  val spark = SparkSession.builder()
    .master("local[6]")
    .appName("transformation")
    .getOrCreate()
  //隐式转换

  import spark.implicits._
  //应用函数
  import org.apache.spark.sql.functions._

  /**
   * 处理的是Nan == >  not a number （也就是double数据类型下的空值）
   */
  @Test
  def nullAndNaN = {
    //创建schema
    val schema = StructType(
      List(
        StructField("id", LongType),
        StructField("year", IntegerType),
        StructField("month", IntegerType),
        StructField("day", IntegerType),
        StructField("hour", IntegerType),
        StructField("session", IntegerType),
        StructField("pm", DoubleType)
      )
    )
    //导入数据集
    val sourceDF: DataFrame = spark.read
      .option("header", true)
      .schema(schema)
      .csv("data/beijingpm_with_nan.csv")
    //丢弃

    // 丢弃的规则
    // 1. ang,只要有一个nan就丢弃
    sourceDF.na.drop("any").show()
    // 2. all，所有数据都是nan的行才丢弃
    sourceDF.na.drop("all").show()
    // 3. 某些列的规则：指定的列按照上述规则
    sourceDF.na.drop("any", List("year", "month", "day", "hour")).show()

    //填充
    //两种规则
    //1.针对所有的数据进行默认值的填充
    sourceDF.na.fill(0).show()
    //2.针对指定列的数据进行默认值的填指
    sourceDF.na.fill(0, List("year", "month")).show()
  }

  /**
   * 处理空值na   字符串数据类型下的空值
   */
  @Test
  def strProcessor() = {
    //读取数据
    val sourceDF = spark.read
      .option("header", true)
      .option("inferSchema", true)
      .csv("data/BeijingPM20100101_20151231.csv")
    //1.丢弃
    sourceDF.where('PM_Dongsi =!= "NA").show()
    //2.替换
    //将PM_Dongsi的的数据类型替换成double类型，将这一列的空值替换成double下的nan
    sourceDF.select(
      'No as "id", 'year, 'month, 'day, 'hour, 'season,
      when('PM_Dongsi === "NA", Double.NaN)
        .otherwise('PM_Dongsi cast (DoubleType))
        .as("pm")
    ).show()

    //replace：使用的基本原则，原类型和转换后的类型必须保持一致
    sourceDF.na.replace("PM_Dongsi", Map("NA" -> "NaN", "NULL" -> "null")).show()
  }
}
